package com.bw.medical.realtime.dwd.app;

import com.bw.medical.realtime.common.base.BaseSQL;
import com.bw.medical.realtime.common.constant.Constant;
import com.bw.medical.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradePrescriptionPaySuc extends BaseSQL {

    public static void main(String[] args) {
        new DwdTradePrescriptionPaySuc().start(
                10014,
                4,
                "medical_dwd_trade_prescription_pay_suc"
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
        readOdsPay(tabEnv,"dwd_trade_prescription_pay");
        readOdsPaySuc(tabEnv,"dwd_trade_prescription_pay_suc");

//        tabEnv.sqlQuery("select * from topic_db_pay_suc where `table` = 'prescription_detail' and `type` = 'update' ").execute().print();


        // 筛选出 处方详情表
        Table presDetail = tabEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['count'] `count`,\n" +
                "`data`['medicine_id'] medicine_id,\n" +
                "`data`['prescription_id'] prescription_id,\n" +
                " et \n" +
                "from topic_db_pay\n" +
                "where `table` = 'prescription_detail'\n" +
                "and `type` = 'insert'");

        tabEnv.createTemporaryView("pre_detail",presDetail);
//        presDetail.execute().print();

        //筛选出支付成功表数据
        Table presDetailPySuc = tabEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['update_time'] update_time,\n" +
                "`data`['total_amount'] total_amount,\n" +
                "`data`['consultation_id'] consultation_id,\n" +
                "`data`['doctor_id'] doctor_id,\n" +
                "`data`['patient_id'] patient_id ,\n" +
                " et  \n" +
                "from topic_db_pay_suc \n" +
                "where `table` = 'prescription'\n" +
                "and `type` = 'update'" +
                "and `data`['status'] = '203' ");
        tabEnv.createTemporaryView("pre_detail_pay_suc",presDetailPySuc);
//        presDetailPySuc.execute().print();

        // 用 Interval Join 进行链接
        Table joinTable = tabEnv.sqlQuery("select\n" +
                "\n" +
                "pd.id,\n" +
                "pps.update_time prescription_pay_suc_time,\n" +
                "total_amount,\n" +
                "consultation_id,\n" +
                "doctor_id,\n" +
                "patient_id,\n" +
                "`count`,\n" +
                "medicine_id,\n" +
                "prescription_id \n" +
                "from \n" +
                "pre_detail as pd,\n" +
                "pre_detail_pay_suc as pps\n" +
                "WHERE pd.prescription_id = pps.id\n" +
                "AND pd.et >= pps.et - INTERVAL '15' minute\n" +
                "AND pd.et <= pps.et - INTERVAL '5' second");

        joinTable.execute().print();

        tabEnv.executeSql("create table "+ Constant.DWD_TRADE_PRESCRIPTION_PAY_SUC+" (\n" +
                "`id` string,\n" +
                "`prescription_pay_suc_time` string,\n" +
                "`count` string,\n" +
                "`medicine_id` string,\n" +
                "`prescription_id` string,\n" +
                "`total_amount` string,\n" +
                "`consultation_id` string,\n" +
                "`doctor_id` string,\n" +
                "`patient_id` string\n" +
                ")" +  SQLUtil.getKafkaSinkSQL(Constant.DWD_TRADE_PRESCRIPTION_PAY_SUC)) ;
//        joinTable.executeInsert(Constant.DWD_TRADE_PRESCRIPTION_PAY_SUC);
    }
}
